package com.bd03.streaminglearn.day0328.test


import com.bd03.streaminglearn.day0327.JedisPoolDemo
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

object Demo1 {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.WARN)
    //配置信息
    val conf: SparkConf = new SparkConf().setAppName("Demo1").setMaster("local[4]")
    //创建Streaming
    val ssc = new StreamingContext(conf, Seconds(5))

    ssc.checkpoint("confirmed")
    ssc.checkpoint("dead")


    //4.统计当前批次的结果数据   新增总人数,新增总死亡    各大洲新增人数,各大洲新增死亡
    val datas: ReceiverInputDStream[String] = ssc.socketTextStream("hdp01", 9999)

    val resData: DStream[(String, String, Int, Int)] = datas.map(t => {
      val arr: Array[String] = t.split(",")
      (arr(0), arr(1), arr(2).toInt, arr(3).toInt)
    })

    resData.map(t => t._3).reduce(_ + _).print()
    resData.map(t => t._4).reduce(_ + _).print()

    resData.map(t => (t._1, t._3)).reduceByKey(_ + _).print()
    resData.map(t => (t._1, t._4)).reduceByKey(_ + _).print()

    //5.统计非中国外的各大洲总人数,总死亡 ,使用redis和updatestateBykey两种方式

    //过滤非中国的信息
    val wai: DStream[(String, String, Int, Int)] = resData.filter(t => t._2 != "中国")
    //一个批次的总人数
    val confirmed: DStream[(String, Int)] = wai.map(t => (t._1, t._3)).reduceByKey(_ + _)
    //一个批次的总死亡
    val dead: DStream[(String, Int)] = wai.map(t => (t._1, t._4)).reduceByKey(_ + _)
    //总人数输出到redis
    confirmed.foreachRDD(t => {
      t.foreachPartition(t => {
        val jedis: Jedis = JedisPoolDemo.getResource()
        t.foreach(t => {
          jedis.hincrBy("confirmed", t._1, t._2)
        })
        jedis.close()
      })
    })
    //总死亡输出到redis
    dead.foreachRDD(t => {
      t.foreachPartition(t => {
        val jedis: Jedis = JedisPoolDemo.getResource()
        t.foreach(t => {
          jedis.hincrBy("dead", t._1, t._2)
        })
        jedis.close()
      })
    })

    //创建函数
    val updateFunc = (value: Seq[Int], count: Option[Int]) => {
      Some(value.sum + count.getOrElse(0))
    }

    //输出结果
    confirmed.updateStateByKey(updateFunc).print()
    dead.updateStateByKey(updateFunc).print()

    //启动
    ssc.start()
    //等待退出
    ssc.awaitTermination()


  }
}
